package com.baishancloud.log.douyin.demand2

import com.baishancloud.log.common.util.{IpInfoUtil, TimeUtil}
import com.baishancloud.log.douyin.mallardName2
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector

import java.util.Objects

/**
 *
 * @author ziqiang.wang 
 * @date 2022/3/1 16:01
 */
class LogParse2 extends ProcessFunction[Origin, Result2] {


  override def processElement(value: Origin, ctx: ProcessFunction[Origin, Result2]#Context, out: Collector[Result2]): Unit = {
    // a 表示需求2中的 a 维度
    var a: String = "-1"
    try {
      val aValue: Array[String] = value.url().split("\\?")(1).split("&")(0).split("=")
      if (aValue.length == 2 && Objects.equals(aValue(0), "a")) {
        a = aValue(1)
      }
    } catch {
      case _: Exception =>
    }
    out.collect(
      Result2(
        TimeUtil.timeAtLong(value.datetimeOfUTC().toEpochSecond, 60),
        60,
        mallardName2,
        "",
        1L,
        Tags(
          value.domain(),
          IpInfoUtil.core(value.ip()).isp,
          a
        )
      )
    )
  }
}
